package net.bwie.realtime.gongdan.dwd.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * 订单详情表
 * @Author: FuHe
 * @Date: 2025/6/10
 * todo 大数据-用户画像-12-达摩盘家庭特征
 */
public class GongDanOrderDeatilJob {
    public static void main(String[] args) {
        // 表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // 输入表  input 映射到kafka消息队列
        createInputTable(tabEnv);
        // 数据处理
        Table reportTable = handle(tabEnv);
        // 输出表  output  映射到clickhouse表
        createOutputTable(tabEnv);
        // 保存数据
        saveToDoris(tabEnv, reportTable);
    }

    private static void saveToDoris(TableEnvironment tabEnv, Table reportTable) {
        tabEnv.createTemporaryView("report_table", reportTable);
        tabEnv.executeSql("INSERT INTO order_sink_clickhouse SELECT * FROM report_table");
    }

    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE order_sink_clickhouse\n" +
                "(\n" +
                "    order_id       STRING COMMENT '订单ID',\n" +
                "    user_id        STRING COMMENT '用户ID',\n" +
                "    item_id        STRING COMMENT '商品ID',\n" +
                "    purchase_time  STRING COMMENT '购买时间',\n" +
                "    quantity       INT COMMENT '购买数量',\n" +
                "    amount         DECIMAL(10, 2) COMMENT '订单金额',\n" +
                "    payment_method STRING COMMENT '支付方式',\n" +
                "    status         int COMMENT '订单状态:1已完成/2已退款'\n" +
                ") WITH (\n" +
                "    'connector' = 'clickhouse',\n" +
                "    'url' = 'jdbc:clickhouse://node103:8123/gongdan12',\n" +
                "    'table' = 'order_transaction',\n" +
                "    'username' = 'default',\n" +
                "    'password' = '',\n" +
                "    'format' = 'json'\n" +
                ")");
    }

    private static Table handle(TableEnvironment tabEnv) {
        Table table = tabEnv.sqlQuery("select * from kafka_order_transaction");
//        table.execute().print();
        return table;
    }

    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE kafka_order_transaction\n" +
                "(\n" +
                "    order_id       STRING COMMENT '订单ID，主键',\n" +
                "    user_id        STRING COMMENT '用户ID',\n" +
                "    item_id        STRING COMMENT '商品ID',\n" +
                "    purchase_time  STRING COMMENT '购买时间',\n" +
                "    quantity       INT COMMENT '购买数量',\n" +
                "    amount         DECIMAL(10, 2) COMMENT '订单金额',\n" +
                "    payment_method STRING COMMENT '支付方式',\n" +
                "    status         int COMMENT '订单状态:1已完成/2已退款'\n" +
                ") WITH (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'order_transaction_topic',           -- Kafka主题名称\n" +
                "    'properties.bootstrap.servers' = 'node101:9092', -- Kafka集群地址\n" +
                "    'properties.group.id' = 'order_transaction_group', -- 消费者组ID\n" +
                "    'format' = 'json',                            -- JSON格式数据\n" +
                "    'json.fail-on-missing-field' = 'false',       -- 允许缺失字段\n" +
                "    'json.ignore-parse-errors' = 'true',          -- 忽略解析错误\n" +
                "    'scan.startup.mode' = 'earliest-offset',        -- 从最新偏移量开始消费\n" +
                "    'properties.auto.offset.reset' = 'latest',\n" +
                "    'format' = 'json',\n" +
                "    'json.fail-on-missing-field' = 'false',\n" +
                "    'json.ignore-parse-errors' = 'true'\n" +
                ")");
    }

    private static TableEnvironment getTableEnv() {
        // 环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        // 时区设置
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        // 并发设置
        configuration.setString("table.exec.resource.default-parallelism", "1");
        // 状态保存时间
        configuration.setString("table.exec.state.ttl", "5s");
        // 返回值
        return tabEnv;
    }
}
